
#include <string.h>
#include <stdio.h>
#include "Entity.h"
#include "Package.h"
#include "MQTTNode.h"
#include "MQTT/MQTT.h"
#include "AiotEntity.h"
#include "MsgHandler.h"
#include "Core/Logger.h"
#include "MqttValidor.h"
#include "MQTT/Params.h"
#include "MQTT/MQTTError.h"
#include "Core/Interface.h"
#include "Core/Utils/Number.h"
#include "Core/Utils/String.h"


static int32_t doConnect(MQTTEntity *handle, ConnectProperty *property);
static int32_t doDisconnect(MQTTEntity *handle, uint8_t reasonCode, DisconnectProperty *property);

static int32_t doUnSubscribe(MQTTEntity *handle, MQTTBuffer *topic, UnSubProperty *prop);
static int32_t doPublish(MQTTEntity *handle, MQTTBuffer *topic, MQTTBuffer *payload, uint8_t qos, PubProperty *prop);
static int32_t doSubscribe(MQTTEntity *handle, MQTTBuffer *topic, RecvHandler handler, uint8_t qos, void *context,SubProperty *prop);

void *UMQTT_CreateEntity(void *context) {

    MQTTEntity *mqttEntity = getSystemInterface()->memMalloc(sizeof(MQTTEntity));
    if (mqttEntity == nullptr) {
        return nullptr;
    }

    memset(mqttEntity, 0, sizeof(MQTTEntity));
    initialAiotEntity(&mqttEntity->aiotEntity,context);

    mqttEntity->cleanSession = 1;
    mqttEntity->rePublishTimeout = 3000;
    mqttEntity->reSubscribeTimeout = 3000;
    mqttEntity->reSendSubscribeEnabled = 0;

    mqttEntity->subMutex = getSystemInterface()->createMutex();
    mqttEntity->pubMutex = getSystemInterface()->createMutex();
    mqttEntity->topicAliasMutex = getSystemInterface()->createMutex();
    mqttEntity->processMutex = getSystemInterface()->createMutex();

    mqttEntity->txPacketSize = 1024 * 1024;

    initialAiotListHead(&mqttEntity->subList);
    initialAiotListHead(&mqttEntity->pubList);
    initialAiotListHead(&mqttEntity->processList);
    initialAiotListHead(&mqttEntity->rxTopicAliasList);
    initialAiotListHead(&mqttEntity->txTopicAliasList);

    mqttEntity->serverReceiveMax = DEFAULT_SERVER_RECEIVE_MAX;

    return mqttEntity;
}

int32_t UMQTT_SetRecvHandle(void *entity, RecvHandler recvHandler) {

    MQTTEntity *mqttEntity = (MQTTEntity *)entity;

    aiotExecCountInc(&mqttEntity->aiotEntity);
    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);

    mqttEntity->recvHandler = recvHandler;

    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
    aiotExecCountDec(&mqttEntity->aiotEntity);

    return aiotSuccess;
}

int32_t UMQTT_SetResendSubscribe(void *entity, uint8_t reSend) {
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;

    aiotExecCountInc(&mqttEntity->aiotEntity);
    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);

    mqttEntity->reSendSubscribeEnabled = reSend;

    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
    aiotExecCountDec(&mqttEntity->aiotEntity);

    return aiotSuccess;
}

int32_t UMQTT_SetCustomInfo(void* entity,MQTTCustomParams* params) {

    MQTTEntity *mqttEntity = (MQTTEntity *)entity;

    aiotExecCountInc(&mqttEntity->aiotEntity);
    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);

    stringDup(&mqttEntity->extendClientId, params->extendClientId);
    cleanMqttSign(mqttEntity);

    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
    aiotExecCountDec(&mqttEntity->aiotEntity);
    return aiotSuccess;
}

int32_t UMQTT_AddTopicHandle(void *entity, char *topic, RecvHandler handle, void *context) {

    int32_t res = aiotSuccess;
//    if (topic == nullptr || entity == nullptr)  {
//        return paramsIsNull;
//    }
//
//    if (strlen(topic) >= MQTT_TOPIC_MAXLEN) {
//        return mqttTopicTooLong;
//    }
//
//    res = mqttCheckTopicValid(topic, strlen(topic));
//    if (res != aiotSuccess) {
//        return res;
//    }
//
//    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
//
//    aiotExecCountInc(&mqttEntity->aiotEntity);
//    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);
//
//    MQTTBuffer    topic_buff;
//    memset(&topic_buff, 0, sizeof(topic_buff));
//    topic_buff.buffer = (uint8_t *)topic;
//    topic_buff.len = strlen(topic);
//
//    getSystemInterface()->enterMutex(mqttEntity->subMutex);
//    res = insertMqttSubList(mqttEntity, &topic_buff, handle, context);
//    getSystemInterface()->leaveMutex(mqttEntity->subMutex);
//
//    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
//    aiotExecCountDec(&mqttEntity->aiotEntity);
    return res;
}

int32_t UMQTT_SetControlInfo(void *entity, MQTTControlParams *params) {

    MQTTEntity *mqttEntity = (MQTTEntity*)entity;
    if (mqttEntity == nullptr || params == nullptr) {
        return paramsIsNull;
    }

    if (mqttEntity->aiotEntity.execEnabled == 0) {
        return disableExecute;
    }

    aiotExecCountInc(&mqttEntity->aiotEntity);
    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);

    mqttEntity->controlInfo = *params;
    memcpy(&mqttEntity->controlInfo,params, sizeof(MQTTControlParams));

    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
    aiotExecCountDec(&mqttEntity->aiotEntity);

    return aiotSuccess;
}

int32_t UMQTT_RemoveTopicHandle(void *entity, char *topic, RecvHandler handle) {

    int32_t res = aiotSuccess;
//    MQTTBuffer    topicBuff = {nullptr,0};
//    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
//    if (topic == nullptr || entity == nullptr) {
//        return paramsIsNull;
//    }
//
//    if (strlen(topic) >= MQTT_TOPIC_MAXLEN) {
//        return mqttTopicTooLong;
//    }
//
//    res = mqttCheckTopicValid(topic, strlen(topic));
//    if (res != aiotSuccess) {
//        return res;
//    }
//
//    aiotExecCountInc(&mqttEntity->aiotEntity);
//    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);
//    memset(&topicBuff, 0, sizeof(topicBuff));
//    topicBuff.buffer = (uint8_t *)topic;
//    topicBuff.len = strlen(topic);
//
//    getSystemInterface()->enterMutex(mqttEntity->subMutex);
//    remoteMqttSubListHandler(mqttEntity, &topicBuff, handle);
//    getSystemInterface()->leaveMutex(mqttEntity->subMutex);
//
//    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
//    aiotExecCountDec(&mqttEntity->aiotEntity);

    return res;
}

int32_t UMQTT_DestroyEntity(void **entity) {

    uint64_t deinitTimeAt = 0;
    MQTTEntity *mqttEntity = nullptr;
    MQTTEvent coreEvent;

    if (entity == nullptr || *entity == nullptr) {
        return paramsIsNull;
    }

    mqttEntity = *(MQTTEntity **)entity;
    if (aiotEnableExecute(&mqttEntity->aiotEntity) == 0) {
        return disableExecute;
    }

    disableAiotExecute(&mqttEntity->aiotEntity);

    SystemInterface* interface = getSystemInterface();
    deinitTimeAt = interface->getTimestamp();

    do {

        if (mqttEntity->aiotEntity.execCount == 0) {
            break;
        }

        interface->sleep(100);
    } while ((interface->getTimestamp() - deinitTimeAt) < mqttEntity->aiotEntity.destroyTimeout);

    if (mqttEntity->aiotEntity.execCount != 0) {
        return mqttExecTimeout;
    }

    remoteTopicAliasList(mqttEntity);

    memset(&coreEvent, 0, sizeof(MQTTEvent));
    coreEvent.type = mqttDestroy;
    handleProcessData(mqttEntity, &coreEvent);

    *entity = nullptr;
    destroyAiotNet(&mqttEntity->aiotEntity);

    deInitialAiotEntity(&mqttEntity->aiotEntity);

    if (mqttEntity->username != nullptr) {
        interface->memFree(mqttEntity->username);
    }

    if (mqttEntity->password != nullptr) {
        interface->memFree(mqttEntity->password);
    }

    if (mqttEntity->clientId != nullptr) {
        interface->memFree(mqttEntity->clientId);
    }

    if (mqttEntity->extendClientId != nullptr) {
        interface->memFree(mqttEntity->extendClientId);
    }

    if (mqttEntity->preConnectProperty != nullptr) {
        interface->memFree(mqttEntity->preConnectProperty);
    }

    interface->destroyMutex(&mqttEntity->subMutex);
    interface->destroyMutex(&mqttEntity->topicAliasMutex);
    interface->destroyMutex(&mqttEntity->pubMutex);
    interface->destroyMutex(&mqttEntity->processMutex);

    destroyMqttSubList(mqttEntity);
    destroyMqttPubList(mqttEntity);
    destroyMqttProcessDataList(mqttEntity);

    interface->memFree(mqttEntity);
    return aiotSuccess;
}

int32_t UMQTT_ProcessReTransport(void *entity) {

    int32_t res = aiotSuccess;
    uint64_t currentAt = 0;
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    AiotEntity* aiotEntity = nullptr;
    if (mqttEntity == nullptr) {
        return paramsIsNull;
    }

    aiotEntity = &mqttEntity->aiotEntity;

    if (aiotEnableExecute(aiotEntity) == 0) {
        return disableExecute;
    }

    aiotExecCountInc(aiotEntity);

    /** mqtt ping packet */
    SystemInterface* interface = getSystemInterface();
    currentAt = interface->getTimestamp();

    if (currentAt < aiotEntity->hblInfo.lastSendAt) {
        aiotEntity->hblInfo.lastSendAt = currentAt;
    }

    if ((currentAt - aiotEntity->hblInfo.lastSendAt) >= aiotEntity->hblInfo.interval) {
        res = sendHeartbeat(mqttEntity);
        if (res == aiotSuccess) {
            aiotEntity->hblInfo.lostTimes++;
        }
        aiotEntity->hblInfo.lastSendAt = currentAt;
    }

    /** mqtt qos1 packet republish */
    reSendPublish(mqttEntity);

    /** mqtt non subscribe ack packet republish */
    if(mqttEntity->reSendSubscribeEnabled) {
        reSendSubscribe(mqttEntity);
    }

    /** mqtt process handler process */
    handleProcessData(mqttEntity, nullptr);
    aiotExecCountDec(aiotEntity);
    return res;
}

int32_t UMQTT_HandleData(void *entity) {

    int32_t res = aiotSuccess,readLen = 0;
    uint32_t mqttRemainLen = 0;
    uint8_t mqttFixedHeader[MQTT_FIXED_HEADER_LEN] = {0};
    uint8_t mqttPktType = 0, mqttPktReserved = 0;
    uint8_t *remain = nullptr;
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    AiotEntity* aiotEntity = nullptr;

    if (mqttEntity == nullptr) {
        return paramsIsNull;
    }

    aiotEntity = &mqttEntity->aiotEntity;
    if (aiotEnableExecute(aiotEntity) == 0) {
        return disableExecute;
    }

    aiotExecCountInc(aiotEntity);

    /** network error reconnect */
    if (aiotEntity->netHandle == nullptr) {
        notifyDisconnectEvent(mqttEntity, networkDisconnect);
    }

    if (aiotEntity->reconnectInfo.enabled == 1 && mqttEntity->disconnectApiCalled == 0) {
        res = sendReconnect(mqttEntity);
        if (res < aiotSuccess) {
            if (res == mqttConnectSuccess) {
                loggerOut(aiotDebug,__FILE__,__LINE__,"MQTT network reconnect success\n");
                notifyConnectEvent(mqttEntity);
                res = aiotSuccess;
            } else {
                aiotExecCountDec(aiotEntity);
                return res;
            }
        }
    }

    /** heartbeat missing reconnect */
    if (aiotEntity->hblInfo.lostTimes > aiotEntity->hblInfo.maxLostTimes) {

        notifyDisconnectEvent(mqttEntity, heartBeatTimeout);
        AiotReconnectInfo* reconnectInfo = &aiotEntity->reconnectInfo;
        uint32_t interval = reconnectInfo->interval;
        uint64_t current = getSystemInterface()->getTimestamp();
        if (current < reconnectInfo->lastRetryTime) {
            reconnectInfo->lastRetryTime = current;
        }
        //重连间隔连接设备
        if (aiotEntity->reconnectInfo.enabled == 1 && mqttEntity->disconnectApiCalled == 0 &&
                current >= (reconnectInfo->lastRetryTime + interval)) {

            loggerOut(aiotTrace,__FILE__,__LINE__,"MQTT heartbeat lost %d times, try to reconnecting...\n",aiotEntity->hblInfo.lostTimes);

            getSystemInterface()->enterMutex(aiotEntity->sendMutex);
            getSystemInterface()->enterMutex(aiotEntity->recvMutex);
            res = handleMqttConnect(mqttEntity, nullptr);
            reconnectInfo->lastRetryTime = getSystemInterface()->getTimestamp();
            getSystemInterface()->leaveMutex(aiotEntity->recvMutex);
            getSystemInterface()->leaveMutex(aiotEntity->sendMutex);

            if (res < aiotSuccess) {
                if (res == mqttConnectSuccess) {
                    loggerOut(aiotDebug,__FILE__,__LINE__,"network reconnect success\n");

                    notifyConnectEvent(mqttEntity);
                    res = aiotSuccess;
                } else {
                    aiotExecCountDec(aiotEntity);
                    return res;
                }
            }
        }
    }

    /** read packet type of mqtt fixed header*/
    getSystemInterface()->enterMutex(aiotEntity->recvMutex);
    res = readMqttMessage(entity, mqttFixedHeader, MQTT_FIXED_HEADER_LEN,&readLen,aiotEntity->recvDataTimeout);
    if (res != aiotSuccess) {
        getSystemInterface()->leaveMutex(aiotEntity->recvMutex);
        aiotExecCountDec(aiotEntity);
        if (res == networkReadLessData) {
            res = aiotSuccess;
        } else {
           // loggerOut(aiotError, __FILE__,__LINE__,"read mqtt message errno:%d!\n", res);
            destroyAiotNet(&mqttEntity->aiotEntity);
        }

        return res;
    }

    /** read remaining packet length of mqtt fixed header */
    res = readRemainlen(mqttEntity, &mqttRemainLen);
    if (res != aiotSuccess) {
        getSystemInterface()->leaveMutex(aiotEntity->recvMutex);
        loggerOut(aiotError, __FILE__,__LINE__,"read remain length errno:%d!\n", res);
        destroyAiotNet(&mqttEntity->aiotEntity);
        aiotExecCountDec(aiotEntity);
        return res;
    }

    /** read remaining bytes */
    res = readRemainbytes(mqttEntity, mqttRemainLen, &remain);
    if (res != aiotSuccess) {

        getSystemInterface()->leaveMutex(aiotEntity->recvMutex);
        loggerOut(aiotError, __FILE__,__LINE__,"read remain bytes errno:%d!\n", res);
        destroyAiotNet(&mqttEntity->aiotEntity);
        aiotExecCountDec(aiotEntity);
        return res;
    }

    getSystemInterface()->leaveMutex(aiotEntity->recvMutex);

    /** get packet type */
    mqttPktType     = mqttFixedHeader[0] & 0xF0;
    mqttPktReserved = mqttFixedHeader[0] & 0x0F;

    /** reset ping response missing times */
    aiotEntity->hblInfo.lostTimes = 0;

    switch (mqttPktType) {
        case MQTT_PINGREQ_PKT_TYPE: {
            res = handlePingRequest(entity);
        }
            break;
        case MQTT_PINGRESP_PKT_TYPE: {
            res = handlePingResponse(mqttEntity, remain, mqttRemainLen);
        }
            break;
        case MQTT_PUBLISH_PKT_TYPE: {
            res = handlerPublish(entity, remain, mqttRemainLen, ((mqttPktReserved >> 1) & 0x03), mqttRemainLen);
        }
            break;
        case MQTT_PUBACK_PKT_TYPE: {
            res = handlerPublishAck(entity, remain, mqttRemainLen);
        }
            break;
        case MQTT_SUBACK_PKT_TYPE: {
            handlerSubUnsubAck(entity, remain, mqttRemainLen, MQTT_SUBACK_PKT_TYPE);
        }
            break;
        case MQTT_UNSUBACK_PKT_TYPE: {
            handlerSubUnsubAck(entity, remain, mqttRemainLen, MQTT_UNSUBACK_PKT_TYPE);
        }
            break;
        case MQTT_SERVER_DISCONNECT_PKT_TYPE: {
            handlerDisconnect(entity, remain, mqttRemainLen);
        }
            break;

        case MQTT_PUBREC_PKT_TYPE:
        case MQTT_PUBREL_PKT_TYPE:
        case MQTT_PUBCOMP_PKT_TYPE: {

        }
            break;
        default: {
            loggerOut(aiotDebug,__FILE__,__LINE__,"recv pkt header:%x\n", mqttPktType);
            res = mqttPacketTypeUnknown;
        }
    }

    if (remain) {
        getSystemInterface()->memFree(remain);
    }

    if (res != aiotSuccess) {
        loggerOut(aiotError, __FILE__,__LINE__,"handle errno:%d!\n", res);
        destroyAiotNet(aiotEntity);
    }

    aiotExecCountDec(aiotEntity);
    return res;
}

int32_t UMQTT_ConnetServer(void *entity, ConnectProperty *property) {

    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    if (mqttEntity == nullptr) {
        return paramsIsNull;
    }

    if (mqttEntity->aiotEntity.registerMode == unknownRegMode) {
        return registerModeNoDefine;
    }

    SystemInterface* interface = getSystemInterface();
    /** clear previous connection properties */
    if (nullptr != mqttEntity->preConnectProperty) {
        interface->memFree(mqttEntity->preConnectProperty);
        mqttEntity->preConnectProperty = nullptr;
    }

    return doConnect(mqttEntity, property);
}

int32_t UMQTT_Publish(void *entity, PublishParams *params) {

    uint8_t appendRid = 0;
    int32_t res = aiotSuccess;
    MQTTBuffer    topicBuff = {0};
    MQTTBuffer    payloadBuff = {0};

    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    if (nullptr == entity || params == nullptr || nullptr == params->topic) {
        return paramsIsNull;
    }

    if (strlen(params->topic) >= MQTT_TOPIC_MAXLEN) {
        return mqttTopicTooLong;
    }

    if (params->payloadLen >= MQTT_PAYLOAD_MAXLEN) {
        return mqttPublishPayloadTooLong;
    }

    memset(&topicBuff, 0, sizeof(topicBuff));
    memset(&payloadBuff, 0, sizeof(payloadBuff));

    appendRid = mqttEntity->appendRequestId;

    SystemInterface* interface = getSystemInterface();
    if (0 == appendRid) {
        topicBuff.buffer = (uint8_t *)params->topic;
        topicBuff.len = strlen(params->topic);
    } else {
        char *ridPrefix = "?_rid=";
        uint64_t timestamp = getSystemInterface()->getTimestamp();
        uint32_t rand = 0;
        uint32_t bufferLen = strlen(params->topic) + strlen(ridPrefix) + 32;
        char *buffer = interface->memMalloc(bufferLen);
        if (nullptr == buffer) {
            return memoryMallocFailed;
        }

        memset(buffer, 0, bufferLen);
        memcpy(buffer,params->topic, strlen(params->topic));
        memcpy(buffer + strlen(buffer), ridPrefix, strlen(ridPrefix));
        uint64ToString(timestamp, buffer + strlen(buffer), nullptr);
        interface->getRand((int8_t *)&rand, sizeof(rand));
        uint32ToString(rand, buffer + strlen(buffer), nullptr);

        topicBuff.buffer = (uint8_t *)buffer;
        topicBuff.len = strlen(buffer);
    }

    payloadBuff.buffer = params->payload;
    payloadBuff.len    = params->payloadLen;

    res = doPublish(mqttEntity, &topicBuff, &payloadBuff, params->qos, params->property);
    if (appendRid != 0) {
        interface->memFree(topicBuff.buffer);
    }

    return res;
}

int32_t UMQTT_Disconnect(void *entity, DisconnectParams *params) {

    int32_t res = aiotSuccess;
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    if (mqttEntity == nullptr) {
        return paramsIsNull;
    }

    if (aiotEnableExecute(&mqttEntity->aiotEntity) == 0) {
        return disableExecute;
    }

    aiotExecCountInc(&mqttEntity->aiotEntity);

    mqttEntity->disconnectApiCalled = 1;

    /** send mqtt disconnect packet to mqtt broker first */
    res = doDisconnect(entity, params->reason, params->property);
    if (res != aiotSuccess) {
        aiotExecCountDec(&mqttEntity->aiotEntity);
        return res;
    }

    remoteTopicAliasList(mqttEntity);

    SystemInterface* interface = getSystemInterface();
    /** close socket connect with mqtt broker */
    interface->enterMutex(mqttEntity->aiotEntity.sendMutex);
    interface->enterMutex(mqttEntity->aiotEntity.recvMutex);
    if (mqttEntity->aiotEntity.netHandle != nullptr) {
        interface->destroyNet(&mqttEntity->aiotEntity.netHandle);
    }

    interface->leaveMutex(mqttEntity->aiotEntity.recvMutex);
    interface->leaveMutex(mqttEntity->aiotEntity.sendMutex);

    notifyDisconnectEvent(mqttEntity, networkDisconnect);

    loggerOut(aiotTrace,__FILE__,__LINE__,"mqtt user calls disconnect api, disconnect\n");

    aiotExecCountDec(&mqttEntity->aiotEntity);

    return res;
}

int32_t UMQTT_Subscribe(void *entity, RecvHandler handler, SubscribeParams *params) {

    MQTTBuffer    topicBuff = {0};
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    if (mqttEntity == nullptr || params == nullptr || params->topic == nullptr) {
        return paramsIsNull;
    }

    if (strlen(params->topic) >= MQTT_TOPIC_MAXLEN) {
        return mqttTopicTooLong;
    }

    memset(&topicBuff, 0, sizeof(topicBuff));

    topicBuff.buffer = (uint8_t *)params->topic;
    topicBuff.len = strlen(params->topic);
    return doSubscribe(mqttEntity, &topicBuff, handler, params->qos, params->context, params->property);
}

int32_t UMQTT_UnSubscribe(void *entity, UnSubscribeParams *params) {

    MQTTBuffer    topicBuff = {0};
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    if (mqttEntity == nullptr || params == nullptr || params->topic == nullptr) {
        return paramsIsNull;
    }

    if (strlen(params->topic) >= MQTT_TOPIC_MAXLEN) {
        return mqttTopicTooLong;
    }

    memset(&topicBuff, 0, sizeof(topicBuff));

    topicBuff.buffer = (uint8_t *)params->topic;
    topicBuff.len = strlen(params->topic);

    return doUnSubscribe(mqttEntity, &topicBuff, params->property);
}

uint8_t UMQTT_SubscribeResult(void *entity, SubscribeParams *params) {
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;
    if (mqttEntity == nullptr) {
        return 0;
    }
    return sendSubResult(mqttEntity);
}

static int32_t doConnect(MQTTEntity *handle, ConnectProperty *property) {

    int32_t res = aiotSuccess;
    uint64_t timeMs = 0;
    SystemInterface* interface = getSystemInterface();
    if (handle == nullptr) {
        return paramsIsNull;
    }

    timeMs = interface->getTimestamp();
    aiotEnableExecute(&handle->aiotEntity);
    aiotExecCountInc(&handle->aiotEntity);

    handle->disconnectApiCalled = 0;

    /** if network connection exist, close first */
    loggerOut(aiotDebug,__FILE__,__LINE__,"[CONNECT]mqtt user calls connect api, connect\n");

    interface->enterMutex(handle->aiotEntity.sendMutex);
    interface->enterMutex(handle->aiotEntity.recvMutex);

    res = handleMqttConnect(handle, property);

    interface->leaveMutex(handle->aiotEntity.recvMutex);
    interface->leaveMutex(handle->aiotEntity.sendMutex);

    if (res == mqttConnectSuccess) {
        uint64_t time_ms = interface->getTimestamp();
        uint32_t time_delta = (uint32_t)(time_ms - timeMs);

        loggerOut(aiotDebug,__FILE__,__LINE__,"[CONNECT]mqtt connect success in %u ms\n",time_delta);

        notifyConnectEvent(handle);
        res = aiotSuccess;
    } else {
        loggerOut(aiotDebug,__FILE__,__LINE__,"[CONNECT]mqtt connect failed\n");
        notifyDisconnectEvent(handle, networkDisconnect);
    }

    aiotExecCountDec(&handle->aiotEntity);
    return res;
}

static int32_t doDisconnect(MQTTEntity *handle, uint8_t reasonCode, DisconnectProperty *property) {

    int32_t res = 0,sendLen = 0;
    uint32_t pktLen = 0;
    uint8_t *pkt = nullptr;
    uint32_t idx = 0;
    MQTTEntity *mqttEntity = (MQTTEntity *)handle;
    uint32_t propertyLen = 0;
    uint32_t propertyLenOffset = 0;
    uint8_t propertyLenArray[4] = {0};
    uint32_t pktLenOffset = 0;
    uint8_t  pktLenArray[4] = {0};
    uint32_t propertyTotalLen = 0;
    uint32_t pktRemainLen = 0;
    GeneralProperty generalProperty = {0};

    loggerOut(aiotDebug,__FILE__,__LINE__,"[DISCONNECT]mqtt user calls disconnect api, disconnect\n");
    if (property != nullptr) {
        memcpy(&(generalProperty.userProperty[0]), &(property->userProperty[0]),MAX_USER_PROPERTY * (sizeof(UserProperty*)));
        generalProperty.reasonString = property->reason;
        int res = mqttGeneralPropertyValid(&generalProperty);
        if (res < aiotSuccess) {
            return res;
        }

        propertyLen = getGeneralPropertyLen(&generalProperty);
    }

    encodeRemainLen(propertyLen, propertyLenArray, &propertyLenOffset);
    propertyTotalLen = propertyLenOffset + propertyLen;

    pktRemainLen = propertyTotalLen + MQTT_V5_DISCONNECT_REASON_CODE_LEN;
    encodeRemainLen(pktRemainLen, pktLenArray, &pktLenOffset);

    SystemInterface* interface = getSystemInterface();
    pktLen =  MQTT_FIXED_HEADER_LEN + pktLenOffset + pktRemainLen ;
    pkt = interface->memMalloc(pktLen);
    if (nullptr == pkt) {
        return memoryMallocFailed;
    }

    memset(pkt, 0, pktLen);

    pkt[idx++] = MQTT_DISCONNECT_PKT_TYPE;
    memcpy(&pkt[idx], pktLenArray, pktLenOffset);
    idx += pktLenOffset;
    pkt[idx++] = reasonCode;
    writeGeneralProp(&pkt[0], &idx, &generalProperty, propertyLenArray, propertyLenOffset);

    interface->enterMutex(mqttEntity->aiotEntity.sendMutex);
    res = writeMqttData(handle, pkt, idx,&sendLen, mqttEntity->aiotEntity.sendDataTimeout);
    interface->leaveMutex(mqttEntity->aiotEntity.sendMutex);
    interface->memFree(pkt);

    loggerOut(aiotDebug,__FILE__,__LINE__,"[DISCONNECT]mqtt disconnect success\n");
    return res;
}

static int32_t doPublish(MQTTEntity *handle, MQTTBuffer *topic, MQTTBuffer *payload, uint8_t qos, PubProperty *prop) {

    int32_t res = aiotSuccess,sendLen = 0;
    uint16_t packetId = 0;
    uint8_t *pkt = nullptr;
    uint32_t idx = 0, remainlen = 0, pkt_len = 0;
    uint16_t aliasId = 0;
    uint8_t replaceTopicWithAlias = 0;
    uint32_t propertyLen = 0;
    uint8_t propertyLenArray[4] = {0};
    uint32_t propertyLenOffset = 0;
    GeneralProperty generalProp = {0};
    SystemInterface* interface = getSystemInterface();

    if (handle == nullptr || payload == nullptr || payload->buffer == nullptr) {
        return paramsIsNull;
    }

    if (topic->len == 0 || qos > MQTT_QOS_MAX) {
        return paramsOutOfRange;
    }

    if (nullptr != prop) {

        memcpy(&(generalProp.userProperty[0]), &(prop->userProperty[0]),MAX_USER_PROPERTY * (sizeof(UserProperty*)));
        generalProp.responseTopic = prop->responseTopic;
        generalProp.correlationData = prop->correlationData;
        generalProp.subIdentifier = prop->subIdentifier;
    }

    res = mqttGeneralPropertyValid(&generalProp);
    if (res < aiotSuccess) {
        return res;
    }

    /** flow control check*/
    res = mqttCheckFlowControl(handle, qos);
    if (res < aiotSuccess) {
        return res;
    }

    /** topic alias*/
    replaceTopicWithAlias = processTxTopicAlias(handle, topic, &aliasId);
    generalProp.topicAlias = aliasId;

    if (aiotEnableExecute(&handle->aiotEntity) == 0) {
        return disableExecute;
    }

    aiotExecCountInc(&handle->aiotEntity);

    if (0 == replaceTopicWithAlias) {
        remainlen = topic->len + payload->len + MQTT_UTF8_STR_EXTRA_LEN;
    } else {
        remainlen = payload->len + MQTT_UTF8_STR_EXTRA_LEN;
    }

    if (qos == MQTT_QOS1) {
        remainlen += MQTT_PACKETID_LEN;
    }

    propertyLen = getGeneralPropertyLen(&generalProp);
    encodeRemainLen(propertyLen, &propertyLenArray[0],&propertyLenOffset);
    remainlen += propertyLen + propertyLenOffset;

    pkt_len = MQTT_FIXED_HEADER_LEN + MQTT_REMAINLEN_MAXLEN + remainlen;
    pkt = interface->memMalloc(pkt_len);
    if (pkt == nullptr) {
        aiotExecCountDec(&handle->aiotEntity);
        return memoryMallocFailed;
    }

    memset(pkt, 0, pkt_len);

    /** Publish Packet Type */
    pkt[idx++] = MQTT_PUBLISH_PKT_TYPE | (qos << 1);

    /** Remaining Length */
    encodeRemainLen(remainlen, &pkt[idx], &idx);

    /** Topic Length */
    if (replaceTopicWithAlias == 0) {
        /** topic len is > 0 when alias DOES NOT replaces topic */
        setUtf8EncodedString((uint8_t *) topic->buffer, topic->len, &pkt[idx]);
        idx += MQTT_UTF8_STR_EXTRA_LEN + topic->len;
    } else {
        /** topic len is 0 when alias replaces topic */
        pkt[idx++] = 0;
        pkt[idx++] = 0;
    }

    /** Packet Id For QOS 1*/
    if (qos == MQTT_QOS1) {
        packetId = getMqttPacketId(handle);
        pkt[idx++] = (uint8_t)((packetId >> 8) & 0x00FF);
        pkt[idx++] = (uint8_t)((packetId) & 0x00FF);
    }

    loggerOut(aiotInfo,__FILE__,__LINE__,"[PUBLISH]mqtt publish topic:%s packetId:%d\n",topic->buffer, packetId);

    writeGeneralProp(pkt, &idx, &generalProp, propertyLenArray, propertyLenOffset);

    /** Payload */
    memcpy(&pkt[idx], payload->buffer, payload->len);
    idx += payload->len;

    pkt_len = idx;

    /** ensure tx pack size does not overflow  */
    res = mqttCheckTXPayloadLen(handle, pkt_len);
    if (res < aiotSuccess) {
        interface->memFree(pkt);
        aiotExecCountDec(&handle->aiotEntity);
        return res;
    }

    if (qos == MQTT_QOS1) {
        interface->enterMutex(handle->pubMutex);
        res = insertMqttPubList(handle, pkt, pkt_len, packetId);
        interface->leaveMutex(handle->pubMutex);
        if (res < aiotSuccess) {
            interface->memFree(pkt);
            aiotExecCountDec(&handle->aiotEntity);
            return res;
        }
    }

    interface->enterMutex(handle->aiotEntity.sendMutex);
    res = writeMqttData(handle, pkt, pkt_len,&sendLen, handle->aiotEntity.sendDataTimeout);
    interface->leaveMutex(handle->aiotEntity.sendMutex);

    if (res < aiotSuccess) {
        interface->memFree(pkt);
        if (res != networkWriteLessData) {
            destroyAiotNet(&handle->aiotEntity);
            loggerOut(aiotError,__FILE__,__LINE__,"[PUBLISH]publish data errno:%d, %p!\n",res, handle->aiotEntity.netHandle);
        }

        aiotExecCountDec(&handle->aiotEntity);
        return res;
    }

    interface->memFree(pkt);
    if (qos == MQTT_QOS1) {
        aiotExecCountDec(&handle->aiotEntity);
        mqttCheckFlowDec(handle);
//        return (int32_t)packetId;
        return aiotSuccess;
    }

    aiotExecCountDec(&handle->aiotEntity);
    return aiotSuccess;
}

static int32_t doUnSubscribe(MQTTEntity *handle, MQTTBuffer *topic, UnSubProperty *prop) {

    int32_t res = aiotSuccess;
    MQTTEntity *mqttEntity = (MQTTEntity *)handle;

    if (mqttEntity == nullptr || topic == nullptr || topic->buffer == nullptr) {
        return paramsIsNull;
    }

    if (topic->len == 0) {
        return paramsOutOfRange;
    }

    if ((res = mqttCheckTopicValid((char *) topic->buffer, topic->len)) < aiotSuccess) {
        return mqttTopicInvalid;
    }

    if (aiotEnableExecute(&handle->aiotEntity) == 0) {
        return disableExecute;
    }

    aiotExecCountInc(&handle->aiotEntity);

    SystemInterface* interface = getSystemInterface();
    interface->enterMutex(mqttEntity->subMutex);
    removeMqttSubList(mqttEntity, topic);
    interface->leaveMutex(mqttEntity->subMutex);

    GeneralProperty genProperty = {0};
    if (nullptr != prop) {
        memcpy(&(genProperty.userProperty[0]), &(prop->userProperty[0]),MAX_USER_PROPERTY * (sizeof(UserProperty*)));
    }

    res = sendSubUnsub(mqttEntity, (char *) topic->buffer, topic->len, 0, MQTT_UNSUB_PKT_TYPE, &genProperty);

    aiotExecCountDec(&mqttEntity->aiotEntity);
    return res;
}

static int32_t doSubscribe(MQTTEntity *handle, MQTTBuffer *topic, RecvHandler handler, uint8_t qos, void *context,
                           SubProperty *prop) {

    int32_t res = aiotSuccess;
    if (handle == nullptr || topic == nullptr || topic->buffer == nullptr) {
        return paramsIsNull;
    }

    if (topic->len == 0 || qos > MQTT_QOS_MAX) {
        return paramsOutOfRange;
    }

    res = mqttCheckTopicValid((char *) topic->buffer, topic->len);
    if (res != aiotSuccess) {
        return mqttTopicInvalid;
    }

    if (aiotEnableExecute(&handle->aiotEntity) == 0) {
        return disableExecute;
    }

    aiotExecCountInc(&handle->aiotEntity);

    SystemInterface* interface = getSystemInterface();
    interface->enterMutex(handle->subMutex);
    res = insertMqttSubList(handle, topic, handler, context);
    interface->leaveMutex(handle->subMutex);
    if (res != aiotSuccess) {
        aiotExecCountDec(&handle->aiotEntity);
        return res;
    }

    GeneralProperty genProperty = {0};
    if (nullptr != prop) {
        memcpy(&(genProperty.userProperty[0]), &(prop->userProperty[0]),MAX_USER_PROPERTY * (sizeof(UserProperty*)));
    }
    /** send subscribe packet */
    res = sendSubUnsub(handle, (char *) topic->buffer, topic->len, qos, MQTT_SUB_PKT_TYPE, &genProperty);
    aiotExecCountDec(&handle->aiotEntity);
    return res;
}

int32_t UMQTT_ClearSign(void *entity) {
    MQTTEntity *mqttEntity = (MQTTEntity *)entity;

    aiotExecCountInc(&mqttEntity->aiotEntity);
    getSystemInterface()->enterMutex(mqttEntity->aiotEntity.dataMutex);

    cleanMqttSign(mqttEntity);

    getSystemInterface()->leaveMutex(mqttEntity->aiotEntity.dataMutex);
    aiotExecCountDec(&mqttEntity->aiotEntity);
    return aiotSuccess;
}